package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_stream_Window1 {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("window")
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val lines = ssc.socketTextStream("localhost",9999)
    val wordOne = lines.map( (_,1))

    /*
        窗口的范围必须是采集周期的整数倍
        第一个参数：窗口时长(一次采集几个采集周期的数据)
        第二个参数：滑块步长（窗口移动几个周期）
     */
    val winDS = wordOne.window(Seconds(6), Seconds(6))

    val wordCount = winDS.reduceByKey(_ + _)

    wordCount.print()

   ssc.start()
   ssc.awaitTermination()
  }

}
